package cn.edu.laud.sany.datatest;

import java.io.BufferedReader;
import java.io.File;
import java.io.FileNotFoundException;
import java.io.FileReader;
import java.io.FileWriter;
import java.io.IOException;
import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.ResultSet;
import java.sql.SQLException;
import java.sql.Statement;
import java.util.List;

import org.apache.cassandra.thrift.Cassandra;
import org.apache.cassandra.thrift.CfDef;
import org.apache.cassandra.thrift.ColumnParent;
import org.apache.cassandra.thrift.ConsistencyLevel;
import org.apache.cassandra.thrift.InvalidRequestException;
import org.apache.cassandra.thrift.KsDef;
import org.apache.cassandra.thrift.NotFoundException;
import org.apache.cassandra.thrift.SlicePredicate;
import org.apache.cassandra.thrift.SliceRange;
import org.apache.cassandra.thrift.TimedOutException;
import org.apache.cassandra.thrift.UnavailableException;
import org.apache.cassandra.utils.ByteBufferUtil;
import org.apache.thrift.TException;
import org.apache.thrift.protocol.TBinaryProtocol;
import org.apache.thrift.protocol.TProtocol;
import org.apache.thrift.transport.TFramedTransport;
import org.apache.thrift.transport.TSocket;
import org.apache.thrift.transport.TTransport;
import org.apache.thrift.transport.TTransportException;

import cn.edu.laud.sany.datatest.RandomKeys.Iterm;
import cn.edu.laud.sany.utils.TimeUtils;

public class BigTableTest
{
	
	/**
	 * @param args
	 * @throws IOException
	 * @throws SQLException
	 * @throws TException
	 * @throws InvalidRequestException
	 * @throws NotFoundException
	 * @throws TimedOutException
	 * @throws UnavailableException
	 */
	public static void main(String[] args) throws NotFoundException,
			InvalidRequestException, TException, SQLException, IOException,
			UnavailableException, TimedOutException
	{
		// TODO Auto-generated method stub
		getAllKeys(args[2], "keys", args);
		// long sum= getTotalColumnNumber(args[2],args);
		// System.out.println(sum);
		// bigTabletest(args[2],"keys/moreThan1000",1000,args);
	}
	
	private static KsDef describeKeyspace(String keyspace, String[] args)
			throws NotFoundException, InvalidRequestException, TException
	{
		TTransport tr = new TFramedTransport(new TSocket(args[0],
				Integer.valueOf(args[1])));
		TProtocol proto = new TBinaryProtocol(tr);
		
		Cassandra.Client client = new Cassandra.Client(proto);
		tr.open();
		KsDef ksDef = client.describe_keyspace(keyspace);
		tr.close();
		return ksDef;
		
	}
	
	public static void getAllKeys(String keyspace, String folderPath,
			String[] args) throws NotFoundException, InvalidRequestException,
			TException, SQLException, IOException
	{
		File folder = new File(folderPath);
		// if(!folder.exists()){
		folder.mkdirs();
		File folderLess10 = new File(folder.getAbsolutePath() + "/less10");
		File folderLess100 = new File(folder.getAbsolutePath() + "/less100");
		File folderLess1000 = new File(folder.getAbsolutePath() + "/less1000");
		File folderMoreThan1000 = new File(folder.getAbsolutePath()
				+ "/moreThan1000");
		File folderMoreThan10000 = new File(folder.getAbsolutePath()
				+ "/moreThan10000");
		folderLess10.mkdir();
		folderLess100.mkdir();
		folderLess1000.mkdir();
		folderMoreThan1000.mkdir();
		// }
		
		{
			try
			{
				Class.forName("org.apache.cassandra.cql.jdbc.CassandraDriver");
			}
			catch (ClassNotFoundException e)
			{
				e.printStackTrace();
			}
		}
		Connection laudConnection = null;
		Statement laudStatement = null;
		try
		{
			laudConnection = DriverManager.getConnection("jdbc:cassandra://"
					+ args[0] + ":" + args[1] + "?version=2.0.0", "", "");
			laudStatement = laudConnection.createStatement();
			laudStatement.execute("use " + keyspace);
		}
		catch (SQLException e1)
		{
			e1.printStackTrace();
		}
		KsDef ksDef = describeKeyspace(keyspace, args);
		for (CfDef cfDef : ksDef.cf_defs)
		{
			long count = 0;
			System.out.println(cfDef.name);
			File file = new File(folder.getAbsolutePath() + "/" + cfDef.name);
			laudStatement.execute("select equip from " + cfDef.name);
			ResultSet rs = laudStatement.getResultSet();
			FileWriter fw = new FileWriter(file);
			while (rs.next())
			{
				fw.write(rs.getString(1) + "\n");
				count++;
			}
			fw.close();
			rs.close();
			if (count > 10000)
			{
				file.renameTo(new File(folderMoreThan10000.getAbsolutePath()
						+ "/" + cfDef.name));
			}
			else if (count > 1000)
			{
				file.renameTo(new File(folderMoreThan1000.getAbsolutePath()
						+ "/" + cfDef.name));
			}
			else if (count > 100)
			{
				file.renameTo(new File(folderLess1000.getAbsolutePath() + "/"
						+ cfDef.name));
			}
			else if (count > 10)
			{
				file.renameTo(new File(folderLess100.getAbsolutePath() + "/"
						+ cfDef.name));
			}
			else
			{
				file.renameTo(new File(folderLess10.getAbsolutePath() + "/"
						+ cfDef.name));
			}
		}
		laudConnection.close();
	}
	
	public static long getTotalColumnNumber(String keyspace, String[] args)
			throws NotFoundException, InvalidRequestException, TException,
			IOException, UnavailableException, TimedOutException
	{
		TTransport tr = new TFramedTransport(new TSocket(args[0],
				Integer.valueOf(args[1])));
		TProtocol proto = new TBinaryProtocol(tr);
		
		Cassandra.Client client = new Cassandra.Client(proto);
		tr.open();
		// KsDef ksDef=client.describe_keyspace(keyspace);
		client.set_keyspace(keyspace);
		File superfolder = new File("keys");
		long sum = 0;
		String maxCf = "";
		long max = 0;
		for (File folder : superfolder.listFiles())
		{
			for (File file : folder.listFiles())
			{
				// if(file.getName().equals("cf_gk_838")){
				System.out.print(".");
				BufferedReader bReader = new BufferedReader(
						new FileReader(file));
				String string = null;
				ColumnParent column_parent = new ColumnParent(file.getName());
				SlicePredicate slicePredicate = new SlicePredicate();
				SliceRange sliceRange = new SliceRange(
						ByteBufferUtil.EMPTY_BYTE_BUFFER,
						ByteBufferUtil.EMPTY_BYTE_BUFFER, false,
						Integer.MAX_VALUE);
				slicePredicate.setSlice_range(sliceRange);
				int count = 0;
				while ((string = bReader.readLine()) != null)
				{
					count = client
							.get_count(ByteBufferUtil.bytes(string),
									column_parent, slicePredicate,
									ConsistencyLevel.ALL);
					sum += count;
					if (count > max)
					{
						max = count;
						maxCf = file.getName();
					}
				}
				
				// }
			}
		}
		System.out.println("maxCf:" + maxCf + ",max size:" + max);
		tr.close();
		return sum;
		
	}
	
	public static void bigTabletest(String keyspace, String folder, int size,
			String[] args) throws SQLException
	{
		{
			try
			{
				Class.forName("org.apache.cassandra.cql.jdbc.CassandraDriver");
			}
			catch (ClassNotFoundException e)
			{
				e.printStackTrace();
			}
		}
		Connection laudConnection = null;
		Statement laudStatement = null;
		try
		{
			laudConnection = DriverManager.getConnection("jdbc:cassandra://"
					+ args[0] + ":" + args[1] + "?version=2.0.0", "", "");
			laudStatement = laudConnection.createStatement();
			laudStatement.execute("use " + keyspace);
		}
		catch (SQLException e1)
		{
			e1.printStackTrace();
		}
		long totalcost = 0;
		List<Iterm> list = RandomKeys.getRandomKeys(folder, size + "");
		System.out.println("begin to query...");
		for (Iterm iterm : list)
		{
			long start = System.nanoTime();
			laudStatement.execute("select * from  " + iterm.cf
					+ " where equip='" + iterm.key + "'");
			ResultSet rSet = laudStatement.getResultSet();
			long cost = System.nanoTime() - start;
			totalcost += cost;
		}
		System.out.println(TimeUtils.nanoSecondToString(totalcost));
		laudStatement.close();
		laudConnection.close();
	}
}
